package main

/**
  @Project: out_mqtt
  @Description:
  @Version 1.0
  @Author: lyndon Create 1.0 8/9/21
  @Copyright (c)2021-2022 gridsum.com
  @Modified By:
*/

import "C"
import (
	"encoding/json"
	"fmt"
	"gitee.com/fluentbit/out_mqtt/config"
	"gitee.com/fluentbit/out_mqtt/pkg/conn"
	"gitee.com/fluentbit/out_mqtt/pkg/util"
	"github.com/fluent/fluent-bit-go/output"
	"github.com/kubeedge/mappers-go/mappers/common"
	"log"
	"reflect"
	"strconv"
	"time"
	"unsafe"
)

//export FLBPluginRegister
func FLBPluginRegister(def unsafe.Pointer) int {
	// Gets called only once when the plugin.so is loaded              加载 plugin.so 时仅调用一次
	return output.FLBPluginRegister(def, "out_mqtt", "Output record to mqtt broker!")
}

//export FLBPluginInit
func FLBPluginInit(plugin unsafe.Pointer) int {
	// Gets called only once for each instance you have configured.   为您配置的每个实例仅调用一次。
	Name := output.FLBPluginConfigKey(plugin, "Name")
	URL := output.FLBPluginConfigKey(plugin, "URL")
	User := output.FLBPluginConfigKey(plugin, "User")
	Passwd := output.FLBPluginConfigKey(plugin, "Passwd")
	Cert := output.FLBPluginConfigKey(plugin, "Cert")
	PrivateKey := output.FLBPluginConfigKey(plugin, "PrivateKey")
	QOS := output.FLBPluginConfigKey(plugin, "QOS")
	Topic := output.FLBPluginConfigKey(plugin, "Topic")
	MsgTag := output.FLBPluginConfigKey(plugin, "MsgTag")
	log.Printf("[out_mqtt] Name = %q", Name)
	log.Printf("[out_mqtt] URL = %q", URL)
	log.Printf("[out_mqtt] User = %q", User)
	log.Printf("[out_mqtt] Passwd = %q", Passwd)
	log.Printf("[out_mqtt] Cert = %q", Cert)
	log.Printf("[out_mqtt] PrivateKey = %q", PrivateKey)
	log.Printf("[out_mqtt] QOS = %q", QOS)
	log.Printf("[out_mqtt] Topic = %q", Topic)
	log.Printf("[out_mqtt] MsgTag = %q", MsgTag)

	qosi, err := strconv.Atoi(QOS)
	if err != nil {
		log.Fatalf("QOS must in the [0 1 2], %v", err)
		return output.FLB_ERROR
	}

	qos := byte(qosi)

	mqttClient := conn.MqttClient{
		MqttClient: common.MqttClient{
			IP:         URL,
			User:       User,
			Passwd:     Passwd,
			Cert:       Cert,
			PrivateKey: PrivateKey,
		},
	}

	if err := mqttClient.StrongConnect(qos); err != nil {
		log.Fatal(fmt.Sprintf("mqtt connect failed, %v\n", err))
	}

	mqttConfig := config.OutMqttConfig{
		Topic:      Topic,
		MqttClient: &mqttClient,
		MsgTag:     MsgTag,
	}

	// Set the context to point to any Go variable
	output.FLBPluginSetContext(plugin, mqttConfig)
	return output.FLB_OK
}

//export FLBPluginFlushCtx
func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int {
	// Gets called with a batch of records to be written to an instance.  使用要写入实例的一批记录调用。
	// Type assert context back into the original type for the Go variable
	mqttConfig := output.FLBPluginGetContext(ctx).(config.OutMqttConfig)
	//log.Printf("flush to topic: %q", mqttConfig.Topic)
	if mqttConfig.MqttClient == nil {
		log.Println("mqttConfig.MqttClient is nil")
	}

	dec := output.NewDecoder(data, int(length))
	count := 0
	for {
		ret, ts, record := output.GetRecord(dec)
		if ret != 0 {
			break
		}

		var timestamp time.Time
		switch t := ts.(type) {
		case output.FLBTime:
			timestamp = ts.(output.FLBTime).Time
		case uint64:
			timestamp = time.Unix(int64(t), 0)
		default:
			log.Println("time provided invalid, defaulting to now.")
			timestamp = time.Now()
		}

		msgContent := make(map[string]interface{})

		var msgSlice []map[string]interface{}
		// Print record keys and values
		msg := make(map[string]interface{})
		msg["count"] = count
		msg["timestamp"] = timestamp.String()
		msg["tag"] = C.GoString(tag)
		for k, v := range record {
			keyOfRcd := util.Interface2String(k)
			ofV := reflect.TypeOf(v)
			if ofV.Kind() == reflect.Slice && ofV.String() == "[]uint8" {
				v = util.UB2S(v.([]uint8))
				//ofV := reflect.TypeOf(v)
				//log.Println("keyOfRcd:", keyOfRcd," v:", v, " type:", ofV.Kind(), " ofStr:", ofV.String())
			}
			msg[keyOfRcd] = v
		}
		count++
		msgSlice = append(msgSlice, msg)

		msgContent[mqttConfig.MsgTag] = msgSlice

		payload, err := json.Marshal(msgContent)
		if err != nil {
			log.Printf("Marshal msg failed, %q", err)
			break
		}

		log.Println("payload:", string(payload))

		err = mqttConfig.MqttClient.Publish(mqttConfig.Topic, payload)
		if err != nil {
			log.Printf("Publish failed, %q", err)
			break
		}
	}
	return output.FLB_OK
}

//export FLBPluginExit
func FLBPluginExit() int {
	return output.FLB_OK
}

func main() {
}
